package cn.edu.thu.laud.thrift;

import java.lang.reflect.Field;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.cassandra.auth.AuthenticatedUser;
import org.apache.cassandra.auth.IAuthenticator;
import org.apache.cassandra.auth.Permission;
import org.apache.cassandra.config.ConfigurationException;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.Int32Type;
import org.apache.cassandra.db.marshal.UTF8Type;
import org.apache.cassandra.thrift.AuthenticationException;
import org.apache.cassandra.thrift.AuthenticationRequest;
import org.apache.cassandra.thrift.AuthorizationException;
import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.thrift.CfDef;
import org.apache.cassandra.thrift.Column;
import org.apache.cassandra.thrift.ColumnDef;
import org.apache.cassandra.thrift.ColumnOrSuperColumn;
import org.apache.cassandra.thrift.ColumnParent;
import org.apache.cassandra.thrift.ColumnPath;
import org.apache.cassandra.thrift.ConsistencyLevel;
import org.apache.cassandra.thrift.InvalidRequestException;
import org.apache.cassandra.thrift.KeyRange;
import org.apache.cassandra.thrift.KeySlice;
import org.apache.cassandra.thrift.KsDef;
import org.apache.cassandra.thrift.NotFoundException;
import org.apache.cassandra.thrift.SchemaDisagreementException;
import org.apache.cassandra.thrift.SlicePredicate;
import org.apache.cassandra.thrift.SliceRange;
import org.apache.cassandra.thrift.TimedOutException;
import org.apache.cassandra.thrift.UnavailableException;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TFramedTransport;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cn.edu.thu.laud.auth.AuthorityGroup;
import cn.edu.thu.laud.auth.MemoryAuthenticator;


public class LaUDAdministrator {
	private static final String PWD_COLUMN = "pwd";
	public static final ByteBuffer PWD_COLUMN_BYTE=UTF8Type.instance.decompose(PWD_COLUMN);
	private static final String LAUD_SYSTEM_CF_KEYSPACES = "kslist";
	private static final String LAUD_SYSTEM_CF_GROUP = "group";
	public static final ByteBuffer LAUD_SYSTEM_COL_GROUP_BYTE=UTF8Type.instance.decompose("group");
	public static final String LAUD_SYSTEM_KS = "LaUD_System_KS";
	private static final String LAUD_SYSTEM_CF_INFO = "info";
	private static Logger logger = LoggerFactory.getLogger(LaUDAdministrator.class);
	public Cassandra.Client daemonClient;
	private TSocket socket;
	LaUDAdministrator(String host,int port,int timeout,String pwd) throws AuthenticationException, AuthorizationException, TException {
		socket = new TSocket(host,port,timeout);
		TTransport transport = new TFramedTransport(socket);
		TProtocol protocol = new TBinaryProtocol(transport);
		daemonClient=new Cassandra.Client(protocol);
		try {
			socket.open();
			Map<String, String> credentials = new HashMap<String, String>();
			/* remove quotes */
			pwd = pwd.replace("\'", "");
			credentials.put(IAuthenticator.USERNAME_KEY, "root");
			credentials.put(IAuthenticator.PASSWORD_KEY, pwd);
			AuthenticationRequest authRequest = new AuthenticationRequest(credentials);
			daemonClient.login(authRequest);
		} catch (AuthenticationException e) {
			logger.error(e.why);
			// TODO Auto-generated catch block
			e.printStackTrace();
			throw e;
		} catch (AuthorizationException e) {
			logger.error(e.why);
			// TODO Auto-generated catch block
			e.printStackTrace();
			throw e;
		} catch (TException e) {
			logger.error(e.getMessage());
			// TODO Auto-generated catch block
			e.printStackTrace();
			throw e;
		}

	}
	public void close(){
		socket.close();
	}
	public void createSystemKeyspaces() throws InvalidRequestException, SchemaDisagreementException, TException{
		KsDef ksDef;
		try {
			ksDef = daemonClient.describe_keyspace(LAUD_SYSTEM_KS);
		} catch (NotFoundException e) {
			logger.warn("not found user keyspace. will create it...");
			//	user's permission group : dbadmin  dbdesigner normaluser and so on...
			List<CfDef> cfdefsList=new ArrayList<CfDef>();
			CfDef group=new CfDef(LAUD_SYSTEM_KS, LAUD_SYSTEM_CF_GROUP);
			group.comparator_type="UTF8Type";
			group.key_validation_class="UTF8Type";
			group.default_validation_class="UTF8Type";
			group.column_metadata=Collections.singletonList(new ColumnDef(LAUD_SYSTEM_COL_GROUP_BYTE, "Int32Type"));
			cfdefsList.add(group);

			//user's info : pwd  mail  sex and so on.
			CfDef info=new CfDef(LAUD_SYSTEM_KS, LAUD_SYSTEM_CF_INFO);
			info.comparator_type="UTF8Type";
			info.key_validation_class="UTF8Type";
			info.default_validation_class="UTF8Type";
			cfdefsList.add(info);

			//the permission of user's specific keyspace/columnfamily such as  user1:ks1/cf1--READ user1:ks1/cf2---ALL 
			CfDef kss=new CfDef(LAUD_SYSTEM_KS, LAUD_SYSTEM_CF_KEYSPACES);
			kss.comparator_type="UTF8Type";
			kss.subcomparator_type="UTF8Type";
			kss.column_type="Super";
			kss.default_validation_class="Int32Type";
			kss.key_validation_class="UTF8Type";
			cfdefsList.add(kss);
			ksDef=new KsDef(LAUD_SYSTEM_KS, "org.apache.cassandra.locator.SimpleStrategy", cfdefsList);
			ksDef.strategy_options=Collections.singletonMap("replication_factor", LaUDLogUtils.parser.getElementString("SystemBackupInfo", "1"));
			String string;
			try {
				string = daemonClient.system_add_keyspace(ksDef);
				logger.debug(string);
				daemonClient.set_keyspace(LAUD_SYSTEM_KS);
				//test TODO
				
				addUser("root", "root");
				addAuthrity(new AuthenticatedUser("root"),AuthorityGroup.DBADMIN);
				addAuthrity(new AuthenticatedUser("root"),AuthorityGroup.SYSTEM);
				
				addUser("testdesign", "test");
				addAuthrity(new AuthenticatedUser("testdesign"),AuthorityGroup.DBDESIGN);
				
				addUser("testadmin", "test");
				addAuthrity(new AuthenticatedUser("testadmin"),AuthorityGroup.DBADMIN);
				
				addUser("testnormal", "test");
				addAuthrity(new AuthenticatedUser("testnormal"),"ks1","cf1",Permission.ALL);
				addAuthrity(new AuthenticatedUser("testnormal"),"ks1","cf2",EnumSet.of(Permission.READ));
				//addAuthrity(new AuthenticatedUser("testnormal"),AuthorityGroup.NORMAL);
			} catch (InvalidRequestException e1) {
				logger.error(e1.getWhy());
				throw e1;
			} catch (SchemaDisagreementException e1) {
				logger.error(e1.getMessage());
				throw e1;
			} catch (TException e1) {
				logger.error(e1.getMessage());
				throw e1;
			}
			logger.debug("add system ks:"+string);

		} catch (InvalidRequestException e) {
			logger.error(e.getWhy());
			throw e;
		} catch (TException e) {
			logger.error(e.getMessage());
			throw e;
		}
	}
	public void addUser(String username,String pwd){
		try {
			ColumnParent parent = new ColumnParent(LAUD_SYSTEM_CF_INFO);
			Column column= new Column(UTF8Type.instance.decompose(PWD_COLUMN)).setValue(UTF8Type.instance.decompose(pwd)).setTimestamp(FBUtilities.timestampMicros());
			daemonClient.insert(UTF8Type.instance.decompose(username), parent, column, ConsistencyLevel.QUORUM);//Ensure that the write has been written to N / 2 + 1 replicas before responding to the client.
		}catch (InvalidRequestException e) {
			logger.error(e.getWhy());
			e.printStackTrace();
		} catch (TException e) {
			logger.error(e.getMessage());
			e.printStackTrace();
		} catch (UnavailableException e) {
			logger.error(e.getMessage());
			e.printStackTrace();
		} catch (TimedOutException e) {
			logger.error(e.getMessage());
			e.printStackTrace();
		}
	}

	public void addAuthrity(AuthenticatedUser user,AuthorityGroup group){
		try {
			ColumnParent parent = new ColumnParent(LAUD_SYSTEM_CF_GROUP);
			int value=0;
			try {
				ColumnOrSuperColumn column=daemonClient.get(UTF8Type.instance.decompose(user.username), new ColumnPath(LAUD_SYSTEM_CF_GROUP).setColumn(LAUD_SYSTEM_COL_GROUP_BYTE), ConsistencyLevel.QUORUM);
				value=Int32Type.instance.compose(column.column.value);
				EnumSet<AuthorityGroup> groups=AuthorityGroup.split(value);
				if(groups.contains(group)){
					return;
				}
			} catch (NotFoundException e) {
				logger.error(e.getMessage());
				//e.printStackTrace();
			}

			value+=group.getValue();
			Column column= new Column(LAUD_SYSTEM_COL_GROUP_BYTE).setValue(Int32Type.instance.decompose(value)).setTimestamp(FBUtilities.timestampMicros());
			daemonClient.insert(UTF8Type.instance.decompose(user.username), parent, column, ConsistencyLevel.QUORUM);//Ensure that the write has been written to N / 2 + 1 replicas before responding to the client.
		}catch (InvalidRequestException e) {
			logger.error(e.getWhy());
			e.printStackTrace();
		} catch (TException e) {
			logger.error(e.getMessage());
			e.printStackTrace();
		} catch (UnavailableException e) {
			logger.error(e.getMessage());
			e.printStackTrace();
		} catch (TimedOutException e) {
			logger.error(e.getMessage());
			e.printStackTrace();
		}
	}
	public void addAuthrity(AuthenticatedUser user,String ks,String cf,EnumSet<Permission> permissions){
		int permission=0;
		if(permissions.size()==2){
			permission=3;
		}else{
			for(Permission p:permissions){
				if(p.equals(Permission.READ))
					permission=1;
				else {
					permission=2;
				}
			}
		}
		try {
			ColumnParent parent = new ColumnParent(LAUD_SYSTEM_CF_KEYSPACES);
			parent.super_column=UTF8Type.instance.decompose(ks);
			Column column= new Column(UTF8Type.instance.decompose(cf)).setValue(Int32Type.instance.decompose(permission)).setTimestamp(FBUtilities.timestampMicros());
			daemonClient.insert(UTF8Type.instance.decompose(user.username), parent, column, ConsistencyLevel.QUORUM);//Ensure that the write has been written to N / 2 + 1 replicas before responding to the client.
		}catch (InvalidRequestException e) {
			logger.error(e.getWhy());
			e.printStackTrace();
		} catch (TException e) {
			logger.error(e.getMessage());
			e.printStackTrace();
		} catch (UnavailableException e) {
			logger.error(e.getMessage());
			e.printStackTrace();
		} catch (TimedOutException e) {
			logger.error(e.getMessage());
			e.printStackTrace();
		}
	}

	/**
	 * 装载所有的用户信息进入内存
	 * load all the user info into memory. this is a brief implementition..
	 * @throws TException 
	 * @throws TimedOutException 
	 * @throws UnavailableException 
	 * @throws InvalidRequestException 
	 */
	public void loadUsers() throws InvalidRequestException, UnavailableException, TimedOutException, TException{
		daemonClient.set_keyspace(LAUD_SYSTEM_KS);
		// read all columns and superColumns
		SlicePredicate predicate = new SlicePredicate();
		SliceRange sliceRange = new SliceRange();
		sliceRange.setStart(new byte[0]).setFinish(new byte[0]);
		sliceRange.setCount(Integer.MAX_VALUE);//column number
		predicate.setSlice_range(sliceRange);
		KeyRange range = new KeyRange(10000);//row number.will allocate java heap size. so we could not 
		ByteBuffer startKey =  ByteBufferUtil.EMPTY_BYTE_BUFFER;
		ByteBuffer endKey =ByteBufferUtil.EMPTY_BYTE_BUFFER ;
		range.setStart_key(startKey).setEnd_key(endKey);


		ColumnParent columnParent = new ColumnParent(LAUD_SYSTEM_CF_INFO);       
		List<KeySlice> userBasicInfos= daemonClient.get_range_slices(columnParent, predicate, range, ConsistencyLevel.QUORUM);

		columnParent = new ColumnParent(LAUD_SYSTEM_CF_KEYSPACES);       
		List<KeySlice> userKsAuthorities= daemonClient.get_range_slices(columnParent, predicate, range, ConsistencyLevel.QUORUM);

		columnParent = new ColumnParent(LAUD_SYSTEM_CF_GROUP);       
		List<KeySlice> userGroupAuthorities= daemonClient.get_range_slices(columnParent, predicate, range, ConsistencyLevel.QUORUM);

		Field field;
		try {
			field = DatabaseDescriptor.class.getDeclaredField("authenticator");
			field.setAccessible(true);
			field.set(null,  FBUtilities.<IAuthenticator>construct(LaUDLogUtils.parser.getElementString("Authenticator", "cn.edu.thu.laud.auth.AllowAllAuthenticator"), "authenticator"));
			field = DatabaseDescriptor.class.getDeclaredField("authority");
			field.setAccessible(true);
			field.set(null,  FBUtilities.<IAuthenticator>construct(LaUDLogUtils.parser.getElementString("Authority", "cn.edu.thu.laud.auth.AllowAllAuthority"), "authority"));
		} catch (SecurityException e) {
			logger.error(e.getMessage());
			e.printStackTrace();
		} catch (NoSuchFieldException e) {
			logger.error(e.getMessage());
			e.printStackTrace();
		} catch (IllegalArgumentException e) {
			logger.error(e.getMessage());
			e.printStackTrace();
		} catch (IllegalAccessException e) {
			logger.error(e.getMessage());
			e.printStackTrace();
		} catch (ConfigurationException e) {
			logger.error(e.getMessage());
			e.printStackTrace();
		}
		MemoryAuthenticator authenticator=(MemoryAuthenticator)DatabaseDescriptor.getAuthenticator();
		AbstractType<String> valueType=UTF8Type.instance;
		//map the users' info.
		for(KeySlice k:userBasicInfos){
			authenticator.userBasicInfos.put(valueType.compose(k.key), k.columns);
		}
		for(KeySlice k:userKsAuthorities){
			authenticator.userKsAuthorities.put(valueType.compose(k.key), k.columns);
		}
		for(KeySlice k:userGroupAuthorities){
			authenticator.userGroupAuthorities.put(valueType.compose(k.key), k.columns);
		}
	}


	public boolean isOpen() {
		return socket.isOpen();
	}



}
